package com.xdja.kafka.hdfs.sink.task;

import com.xdja.kafka.constants.GlobalConstant;
import com.xdja.kafka.hdfs.sink.HdfsSinkConnector;
import com.xdja.kafka.hdfs.sink.handle.ParquetSinkHandle;
import com.xdja.kafka.hdfs.sink.writer.definition.WriterDefinition;
import com.xdja.kafka.hdfs.sink.writer.manage.ParquetWriterManage;
import com.xdja.kafka.util.LoginUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Collection;
import java.util.Map;
import java.util.Properties;

/**
 * 向hdfs写入parquet格式数据的任务
 */

public class ParquetSinkTask extends SinkTask{
    private static final Logger log = LoggerFactory.getLogger(ParquetSinkTask.class);
    /**
     * 已经flush的次数
     */
    private int haveFlushCounter = 0;

    @Override
    public String version() {
        String version = new HdfsSinkConnector().version();
        log.info("版本为:{}", version);
        return version;
    }

    @Override
    public void start(Map<String, String> props) {
        log.info("开始执行task的start()方法");
        /*1.kafka连接hadoop基本配置信息*/
        WriterDefinition.hdfsUrl = props.get(GlobalConstant.HDFS_URL);
        WriterDefinition.hdfsPath = props.get(GlobalConstant.HDFS_PATH);
        log.info("获取到hadoop的相关配置， hdfsUrl:{}, hdfsPath:{}", WriterDefinition.hdfsUrl, WriterDefinition.hdfsPath);
        Configuration configuration = new Configuration();
        //开启支持FileSystem支持append()方法。并且需要hadoop非安全模式，linux命令行开启：hadoop dfsadmin -safemode leave
        configuration.setBoolean("dfs.support.append", true);
        configuration.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
        //指定hdfs驱动处理类
        configuration.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        //必须有此配置，否则默认开启的LocalFileSystem，从而不支持ileSystem支持append()方法
        configuration.set("fs.defaultFS", WriterDefinition.hdfsUrl);
        //华为云认证的需要指定hadoop的配置文件
        String hdfsLoginType = props.get(GlobalConstant.HDFS_LOGIN_TYPE);
        if(!"0".equals(hdfsLoginType)) {
            configuration.addResource(new Path( props.get(GlobalConstant.HUAWEI_HDFS_SITE_PATH)));
            configuration.addResource(new Path( props.get(GlobalConstant.HUAWEI_CORE_SITE_PATH)));
        }

        /*2.系统配置*/
        Properties properties = System.getProperties();
        //hadoop访问时的用户为root
        properties.setProperty("HADOOP_USER_NAME", "root");

        /*3.创建根目录*/
        //文件所在根目录
        Path rootHdfsPath = new Path(WriterDefinition.hdfsPath);
        try{
            WriterDefinition.fileSystem = FileSystem.get(configuration);
            if(!WriterDefinition.fileSystem.exists(rootHdfsPath)) {
                //创建根目录，并赋予777权限
                WriterDefinition.fileSystem.mkdirs(rootHdfsPath, WriterDefinition.fsPermission_777);
                log.info("根目录不存在，创建根目录.hdfsPath:{}", WriterDefinition.hdfsPath);
            }
        }catch (IOException e){
            //发生异常时，直接退出
            this.stop();
            log.error("创建hdfs的输出流失败。错误原因:{}", e.getMessage(), e);
        }

    }

    @Override
    public void put(Collection<SinkRecord> sinkRecords) {
        log.info("开始执行task的put()方法， 读取到的记录数为:{}", sinkRecords.size());
        try {
            ParquetSinkHandle parquetSinkRecordHandle = new ParquetSinkHandle(sinkRecords);
            parquetSinkRecordHandle.doHandle();
        } catch (Throwable e) {
            log.error("保存记录失败:{}", e.getMessage(), e);
            //发生异常时关闭该connector
            this.stop();
        }
    }

    @Override
    public void flush(Map<TopicPartition, OffsetAndMetadata> offsets) {
        //flush指定次数后，进行华为云认证一次
        haveFlushCounter ++;
        if(haveFlushCounter % HdfsSinkConnector.connectHdfsConfig.getHuaweiLoginBean().getFlushCounter() == 0) {
            try {
                log.warn("开始进行定期的华为云认证，已flush次数：{}", haveFlushCounter);
                LoginUtil.login(HdfsSinkConnector.connectHdfsConfig.getHuaweiLoginBean());
            } catch (IOException e) {
                log.error("华为云hadoop认证失败，终止本connector运行, 原因：{}", e.getMessage(), e);
                throw new RuntimeException("华为云hadoop认证失败");
            }
        }
    }

    @Override
    public void stop() {
        log.info("开始执行task的stop()方法");
        /*1.关闭已缓存的ParquetWriter*/
        ParquetWriterManage.getInstance().clearWriterMap();
        log.info("已清空ParquetWrite集合");
        /*2.关闭fileSystem*/
        try {
            if (WriterDefinition.fileSystem != null) {
                WriterDefinition.fileSystem.close();
                log.info("成功关闭fileSystem");
            }
        } catch (IOException e) {
            log.error("stop(), 关闭fileSystem失败。错误原因：{}", e.getMessage(),  e);
        }
    }
}
